package com.bigdata.spark.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author Gerry chan
 * @version 1.0
 *  2020/12/28 21:33
 *  需求：通过 SparkStreaming 从 Kafka 读取数据，并将读取过来的数据做简单计算，最终打印
 *  到控制台。
 */
object SparkStreaming04_Kafka {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("sparkStreaming-kafka")

    val ssc = new StreamingContext(sparkConf, Seconds(3))

    //定义kafka参数
    val kafkaPara:Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux1:9092,linux2:9092,linux3:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )

    //读取Kafka数据创建DStream
    val kafkaDStream:InputDStream[ConsumerRecord[String, String]]  =
      KafkaUtils.createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Set("atguigu"), kafkaPara)
      )

    //将每条消息的KV取出
    val valueDStream:DStream[String] = kafkaDStream.map(record => record.value())

    //计算WordCount
    valueDStream.flatMap(_.split(" "))
        .map((_,1))
        .reduceByKey(_+_)
        .print()

    ssc.start()
    ssc.awaitTermination()
  }

}
